package biz

import (
	"context"
	"fmt"
	v1 "kratos_kafka/api/helloworld/v1"
	"kratos_kafka/internal/broker"
	"kratos_kafka/internal/conf"
	"kratos_kafka/internal/utils/mq_kafka"
	"time"

	"github.com/go-kratos/kratos/v2/errors"
	"github.com/go-kratos/kratos/v2/log"
)

var (
	// ErrUserNotFound is user not found.
	ErrUserNotFound = errors.NotFound(v1.ErrorReason_USER_NOT_FOUND.String(), "user not found")
)

// Greeter is a Greeter model.
type Greeter struct {
	Hello string
}

// GreeterRepo is a Greater repo.
type GreeterRepo interface {
	Save(context.Context, *Greeter) (*Greeter, error)
}

// GreeterUsecase is a Greeter usecase.
type GreeterUsecase struct {
	repo GreeterRepo
	log  *log.Helper
	// Notice broker
	broker *broker.Broker
	// Notice 项目所有的yaml配置项都放在这里
	yamlConf *conf.Bootstrap
}

// NewGreeterUsecase new a Greeter usecase.
func NewGreeterUsecase(repo GreeterRepo, logger log.Logger, broker *broker.Broker, yamlConf *conf.Bootstrap) *GreeterUsecase {
	return &GreeterUsecase{
		repo:     repo,
		log:      log.NewHelper(logger),
		broker:   broker,
		yamlConf: yamlConf,
	}
}

func (uc *GreeterUsecase) CreateGreeter(ctx context.Context, req *v1.SayHelloReq) error {
	currArticle := mq_kafka.ArticleData{
		Title:       req.Title,
		Desc:        req.Desc,
		Tags:        req.Tags,
		PublishTime: time.Now(),
	}
	// Notice 直接传结构体数据即可
	errPublish := uc.broker.Kafka.Publish(uc.yamlConf.MessageQueue.Kafka.ArticleTopic, currArticle)
	if errPublish != nil {
		fmt.Println("往kafka中发送数据出现错误了！！", errPublish)
	} else {
		fmt.Println("成功往kafka中写入了数据!", currArticle)
	}

	return nil
}

// 可以做一些业务逻辑，接收从kafka中收到的数据并处理
func (uc *GreeterUsecase) AutoHandleArticleData(ctx context.Context, msg *mq_kafka.ArticleData) error {

	if msg == nil || msg.Title == "" || msg.Tags == nil || len(msg.Tags) < 1 {
		log.Warn("AutoHandleArticleData kafka中ArticleData的数据不正确！", msg.Title, msg.Tags)
		return nil
	}

	fmt.Println("从kafka中接收到正确的articleData数据: ", msg.Title, msg.Desc, msg.Tags, msg.PublishTime)

	/// Notice 老版本的写法：往kafka中发数据～后面可以为每一次发送数据设置相关参数，
	//msg.Desc = "往kafka中发送数据的desc" // 往kafka中发送的数据的desc改成不一样的
	//body, errBody := json.Marshal(msg)
	//if errBody != nil {
	//	return errBody
	//}
	////  WithBatchSize: 默认100条发送一次，设置成1的话表示有一条发一条！！！
	////  WithAsync: 异步发送
	//errPublish := uc.broker.Kafka.Publish(uc.yamlConf.MessageQueue.Kafka.ArticleTopic, body, kafka.WithBatchSize(1), kafka.WithAsync(true))
	//if errPublish != nil {
	//	fmt.Println("往kafka中发送数据出现错误了！！", errPublish)
	//}
	return nil
}
